package com.bunnymaicai.boot.kafka.autoconfigure;

import com.bunnymaicai.boot.kafka.autoconfigure.model.KafkaConfiguration;
import org.apache.kafka.clients.producer.Producer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;

/**
 * 描述信息
 *
 * @author youshang
 * @date 2021/2/25 16:22
 */
@Configuration
@EnableConfigurationProperties
public class KafkaAutoConfiguration {

    @Bean
    @ConditionalOnMissingBean
    @ConfigurationProperties("spring.kafka")
    public KafkaConfiguration kafkaConfiguration() {
        return new KafkaConfiguration();
    }

    @Bean
    @ConditionalOnMissingBean
    public Producer producer(KafkaConfiguration kafkaConfiguration) {
        Properties properties = new Properties();
        properties.put(org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfiguration.getBootstrapServers());
        properties.put(org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, kafkaConfiguration.getProducer().getKeySerializer());
        properties.put(org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, kafkaConfiguration.getProducer().getValueSerializer());
        org.apache.kafka.clients.producer.Producer<Object, Object> producer = new org.apache.kafka.clients.producer.KafkaProducer<>(properties);
        return producer;
    }

}
